package com.kitten.count.consumer;

import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Maps;
import com.kitten.count.constant.MQConstants;
import com.kitten.count.constant.RedisKeyConstants;
import com.kitten.count.enums.CollectUnCollectTypeEnum;
import com.kitten.count.model.dto.AggregationCountCollectUnCollectNoteMqDTO;
import com.kitten.count.model.dto.AggregationCountLikeUnlikeNoteMqDTO;
import com.kitten.count.model.dto.CountCollectUnCollectMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kittenBook_group_" + MQConstants.TOPIC_COUNT_NOTE_COLLECT,
    topic = MQConstants.TOPIC_COUNT_NOTE_COLLECT
)
public class CountNoteCollectConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    //Buffer Trigger
    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000)  // 缓存50000条
            .batchSize(1000)    // 每1000条聚合一次
            .linger(Duration.ofSeconds(5))  // 5s聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();

    @Override
    public void onMessage(String body) {
        bufferTrigger.enqueue(body);
    }

    private void consumeMessage(List<String> messages) {
        log.info("聚合 received messages: {}, size: {}", messages, messages.size());
        List<CountCollectUnCollectMqDTO> dtos = messages.stream().map(message -> JsonUtils.parseObject(message, CountCollectUnCollectMqDTO.class)).toList();

        Map<Long, List<CountCollectUnCollectMqDTO>> noteId2CountCollectMap = dtos.stream().collect(Collectors.groupingBy(CountCollectUnCollectMqDTO::getNoteId));

        List<AggregationCountCollectUnCollectNoteMqDTO> countList = null;

        for(Map.Entry<Long, List<CountCollectUnCollectMqDTO>> entry : noteId2CountCollectMap.entrySet()) {
            Long noteId = entry.getKey();
            List<CountCollectUnCollectMqDTO> value = entry.getValue();
            Long creatorId = null;
            int finalCount = 0;
            for(CountCollectUnCollectMqDTO dto : value) {
                Integer type = dto.getType();
                creatorId = dto.getNoteCreatorId();
                CollectUnCollectTypeEnum collectUnCollectTypeEnum = CollectUnCollectTypeEnum.valueOf(type);
                if (collectUnCollectTypeEnum == null) {
                    continue;
                }
                switch (collectUnCollectTypeEnum) {
                    case COLLECT -> {
                        finalCount++;
                    }
                    case UNCOLLECT -> {
                        finalCount--;
                    }
                }
            }
            countList.add(AggregationCountCollectUnCollectNoteMqDTO.builder()
                    .noteId(noteId)
                    .creatorId(creatorId)
                    .count(finalCount)
                    .build()
            );
        }


        // 更新 Redis
        countList.forEach(countItem -> {
            // 笔记发布者 ID
            Long creatorId = countItem.getCreatorId();
            // 笔记 ID
            Long noteId = countItem.getNoteId();
            // 聚合后的计数
            Integer count = countItem.getCount();

            String countNoteRedisKey = RedisKeyConstants.buildCountNoteKey(noteId);     // count:note:noteId
            Boolean isCountNoteExists = redisTemplate.hasKey(countNoteRedisKey);
            if (isCountNoteExists) {
                redisTemplate.opsForHash().increment(countNoteRedisKey, RedisKeyConstants.FIELD_COLLECTION_TOTAL, count); // key, hashKey, delta
            }
            // 更新 Redis 用户维度点赞数
            String countUserRedisKey = RedisKeyConstants.buildCountUserKey(creatorId);  // count:user:userId
            Boolean isCountUserExists = redisTemplate.hasKey(countUserRedisKey);
            if (isCountUserExists) {
                redisTemplate.opsForHash().increment(countUserRedisKey, RedisKeyConstants.FIELD_LIKE_TOTAL, count);
            }

        });

        // MQ: 笔记收藏数据落库
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countList)).build();
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COLLECT_2_DB, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("笔记收藏数据落库 消息发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("笔记收藏数据 消息发送失败", throwable);
            }
        });


    }


}
